%%---------------------------------------------------------
%% Author: vvorobyov
%% Created: 09.04.2009
%% Description: TODO: Add description to valg
%%---------------------------------------------------------

-module(valg).

%%---------------------------------------------------------
%% Include files
%%---------------------------------------------------------

-include("../include/valg.hrl").
-include("../include/rrs.hrl").
-include("../include/pipe.hrl").
-include("../include/iter.hrl").
-include("../include/pgt.hrl").
-include("../include/util.hrl").

%%---------------------------------------------------------
%% Exported Functions
%%---------------------------------------------------------
-export([
		 expr/6,
		 eval/2,
		 explain/1,
		 test/1,
		 load_ages_hobbies/0]).

%%%%%%%%%%%%%%%%%%%%%%%%%%%% V-Algebra %%%%%%%%%%%%%%%%%%%%%%%%%%
%% 
%% RS			-	LOG|PORT|COLLECT|{'$rs-filter',RS,FILTER_SPEC}			:ITER|PIPE
%% VECT			-	JOIN|MAP|FILTER|KEY_FOLD|VFILTER
%% VECTs		-	'['VECT [,VECTs]']'       
%% 
%% RSFILTER		-	{'$filter',RS,FILTER_SPEC}
%% RSFORMAT		-	{'$format',RS,FORMAT_SPEC}
%% 
%% JOIN			-	{'$join',VECTs[,VECT %Destination vect%]}				:VECT
%% MAP			-	{'$map',VECT|RS[,VECT|VFORMAT %Destination vect%]}    	:VECT
%% VFILTER		-	{'$filter',VECT,FILTER_SPEC}							:VECT
%% KEY_FOLD		-	{'$key-fold',VECT,FOLD_SPEC}							:VECT
%% COLLECT		-	{'$collect',VECT[,NODE,RS %Destination RS%]}			:RS
%% VFORMAT		-	{'$format',VECT,FORMAT_SPEC}
%% 
%% KEY_SPEC		-	{ATTR_NUM|FUNC}
%% FILTER_SPEC 	-	{FUNC|MS}
%% FOLD_SPEC	-	'$count'|
%% 					{'$max',ATTR_NUM}|
%% 					{'$min',ATTR_NUM}|
%% 					{'$avg',ATTR_NUM}|
%% 					FUNC_FOLD
%% 
%% PORT			-	{'$src-port',PORT_SPEC}									:ITER|PIPE
%% LOG			-	{'$src-log',LOG_NAME}									:ITER|PIPE 

%%---------------------------------------------------------
%% API Functions
%%---------------------------------------------------------

eval(Expr,OutPipe) ->
	MainPg={pg,make_ref(),"Evaluate expression"},
	?LOG_TRACE("NORM EXPR ~p~n", [expr_norm(Expr)]),
	PgtDict=expr(expr_norm(Expr),_ExprPath=[],[OutPipe],MainPg,dict:new(),make_ref()),
	#pg{
		id=erlang:make_ref(),
		name="Controller",
		pgs=[make_pgt(PgtDict,MainPg)],
		workers=[]
	   }.

expr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id) 
  when element(1,VectExpr)==?VECT_TAG ->
	vexpr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id);
expr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id) 
  when element(1,VectExpr)==?JOIN_TAG ->
	jexpr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id);
expr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id) 
  when element(1,VectExpr)==?JOIN_LKP_TAG ->
	jlkpexpr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id);
expr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id) 
  when element(1,VectExpr)==?MAP_TAG ->
	mexpr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id);
expr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id) 
  when element(1,VectExpr)==?FILTER_TAG ->
	fexpr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id);
expr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id) 
  when element(1,VectExpr)==?KEYFOLD_TAG ->
	kfexpr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id);
expr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id) 
  when element(1,VectExpr)==?FORMAT_TAG ->
	fmtexpr(VectExpr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id).

expr_norm(VectExpr) 
  when element(1,VectExpr)==?JOIN_TAG ->
	_VectExprNorm=erlang:setelement(2, VectExpr, [norm_jarg(XExpr)||XExpr<-element(2,VectExpr)]);
expr_norm(VectExpr) ->
	VectExpr.

norm_jarg(Expr) when element(1,Expr)/=?MAP_TAG ->
	{?MAP_TAG,expr_norm(Expr),fun(X)-> X end};
norm_jarg(Expr) ->
	Expr.

jexpr({?JOIN_TAG,VectExprList,RecFun}=Expr,ExprPath,[ToPipe|_Other]=PipeSequence,ParentPg,PgtDict,Id) ->
	JoinSpecAndVectExprList=[{_ExprId=erlang:make_ref(),_KeyFun=not_provided_key_fun,VectExpr}||VectExpr<-VectExprList],
	JoinSpecs=[{ExprId,KeyFun}||{ExprId,KeyFun,_VectExpr}<-JoinSpecAndVectExprList],
	FormatPipe=pipe_mapper:new(RecFun, ToPipe),
	JPipe=pipe_join:new(JoinSpecs, FormatPipe),
	NewPg={pg,Id,"Join"},
	PgtDictNew=dict:append(ParentPg, NewPg , PgtDict),
	lists:foldl(fun({ExprId,_KeyFun,VectExpr},PgtDictAcc)->
					expr(VectExpr,[Expr|ExprPath],[JPipe|PipeSequence],NewPg,PgtDictAcc,ExprId)
				end	, PgtDictNew, JoinSpecAndVectExprList).

mexpr({?MAP_TAG,RsExpr,RecFun}=Expr,ExprPath,[ToPipe|_Other]=PipeSequence,ParentPg,PgtDictAcc,Id) 
  	when element(1,RsExpr)==?RS_TAG ->
	MapredPipe=pipe_mapred:new(Id, strip_sender_sign(ExprPath,ToPipe)),
	FormatPipe=pipe_mapper:new(RecFun, MapredPipe),
	rsexpr(RsExpr,[Expr|ExprPath],[FormatPipe|PipeSequence],ParentPg,PgtDictAcc,erlang:make_ref());
mexpr({?MAP_TAG,VectExpr,RecFun}=Expr,ExprPath,[ToPipe|_Other]=PipeSequence,ParentPg,PgtDictAcc,Id) ->
	MapredPipe=pipe_mapred:new(Id, strip_sender_sign(ExprPath,ToPipe)),
	FormatPipe=pipe_mapper:new(RecFun, MapredPipe),
	expr(VectExpr,[Expr|ExprPath],[FormatPipe|PipeSequence],ParentPg,PgtDictAcc,erlang:make_ref()).
	
fmtexpr({?FORMAT_TAG,VExpr,RecFun}=_Expr,ExprPath,[ToPipe|_Other]=PipeSequence,ParentPg,PgtDictAcc,Id) when is_function(RecFun) ->
	FormatPipe=pipe_mapper:new(RecFun, ToPipe),
	expr(VExpr,ExprPath,[FormatPipe|PipeSequence],ParentPg,PgtDictAcc,Id).

vexpr({?VECT_TAG,V,RecFun}=_Expr,ExprPath,[ToPipe|_Other]=PipeSequence,ParentPg,PgtDictAcc,Id) when is_function(RecFun) ->
	FormatPipe=pipe_mapper:new(RecFun, ToPipe),
	vexpr({?VECT_TAG,V},ExprPath,[FormatPipe|PipeSequence],ParentPg,PgtDictAcc,Id);
vexpr({?VECT_TAG,V}=_Expr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id) ->
	vexpr({?VECT_TAG,V,all},ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id);
vexpr({?VECT_TAG,V,{lookup,_Key}=Query,RecFun}=_Expr,ExprPath,[ToPipe|_Other]=PipeSequence,ParentPg,PgtDictAcc,Id) when is_function(RecFun)->
	FormatPipe=pipe_mapper:new(RecFun, ToPipe),
	vexpr({?VECT_TAG,V,Query},ExprPath,[FormatPipe|PipeSequence],ParentPg,PgtDictAcc,Id);
vexpr({?VECT_TAG,V,{lookup,Key}=Query}=Expr,ExprPath,[ToPipe|_Other]=_PipeSequence,ParentPg,PgtDictAcc,Id) ->
	Node=?ROUTE2UPNODE(svc_rrs:key_route(Key)),
	Iter=iter_vect_query:new(V, Query),
 	vpump_work(Node,Iter,[Expr|ExprPath],
	ToPipe,ParentPg,PgtDictAcc,Id);
vexpr({?VECT_TAG,V,all}=Expr,ExprPath,[ToPipe|_Other]=_PipeSequence,ParentPg,PgtDictAcc,_Id) ->
	Snap=svc_rrs:make_route_snapshot(),
	UpNodes=route_snap_to_nodes(Snap),% TODO Snapshot must be done once and globally to have entire pgt plan builded in context of one cluster state
	Iter=iter_vect_query:new(V, all),
	lists:foldl(fun(Node,PgtDictAccIn)->
						vpump_work(Node,Iter,[Expr|ExprPath],
							ToPipe,
							ParentPg,PgtDictAccIn,make_ref())
				end, PgtDictAcc, UpNodes).

fexpr({?FILTER_TAG,VExpr,FiltPredFun}=Expr,ExprPath,[ToPipe|_Other]=PipeSequence,ParentPg,PgtDictAcc,Id) ->
	FiltPipe=pipe_filter:new(FiltPredFun, ToPipe),
	expr(VExpr,[Expr|ExprPath],[FiltPipe|PipeSequence],ParentPg,PgtDictAcc,Id).

kfexpr({?KEYFOLD_TAG,VExpr,#kfm{fold_fun=FoldFun,acc0=Acc0,record_fun=RecFun}}=_Expr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id) ->
%% 	?LOG_TRACE("KFERPR ~p!!!!!!!!!!!!",[Expr]),
	kfexpr({?KEYFOLD_TAG,VExpr,FoldFun,Acc0,RecFun},ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id);
kfexpr({?KEYFOLD_TAG,VExpr,FoldFun,Acc0,RecFun}=Expr,ExprPath,[ToPipe|_Other]=PipeSequence,ParentPg,PgtDictAcc,Id) ->
	KFPipe=pipe_key_fold:new({FoldFun,Acc0,RecFun},ToPipe),
	expr(VExpr,[Expr|ExprPath],[KFPipe|PipeSequence],ParentPg,PgtDictAcc,Id).
jlkpexpr({?JOIN_LKP_TAG,VExpr,VectSpecList,RecFun}=Expr,ExprPath,[ToPipe|_Other]=PipeSequence,ParentPg,PgtDictAcc,Id) ->
	NewPg={pg,Id,"Join lookup"},
	PgtDictNew=dict:append(ParentPg, NewPg , PgtDictAcc),
	FormatPipe=pipe_mapper:new(RecFun, ToPipe),
	LookupFunList=[
				   begin
					   case VectSpec of
						   {?VECT_TAG,V} ->
							   fun(Record) ->
									   ?LOG_TRACE("LOOKUP FUN1 ~p~n", [Record]),
									   Key=element(2,Record),
									   iter_vect_query:new(V, {lookup,Key})
							   end;
						   {?VECT_TAG,V,VRecFun} when is_function(VRecFun,1) ->
							   fun(Record) ->
									   ?LOG_TRACE("LOOKUP FUN2 ~p [~p,~p]~n", [Record,node(),self()]),
									   Key=element(2,Record),
									   BaseIter=iter_vect_query:new(V, {lookup,Key}),
									   iter_mapper:new(BaseIter, VRecFun)
							   end;
						   {?FILTER_TAG,{?VECT_TAG,V},FiltPredFun} ->
							   fun(Record) ->
									   ?LOG_TRACE("LOOKUP FUN3 ~p~n", [Record]),
									   Key=element(2,Record),
									   iter_filter:new(iter_vect_query:new(V, {lookup,Key}), FiltPredFun)
							   end;
						   {?FILTER_TAG,{?VECT_TAG,V,VRecFun},FiltPredFun} when is_function(VRecFun,1) ->
							   fun(Record) ->
									   ?LOG_TRACE("LOOKUP FUN4 ~p~n", [Record]),
									   Key=element(2,Record),
									   BaseIter=iter_vect_query:new(V, {lookup,Key}),
									   iter_filter:new(iter_mapper:new(BaseIter, VRecFun), FiltPredFun)
							   end;
						   BadVectSpec ->
							   io:format("VECT SPECS ~p~n", [VectSpecList]),
							   exit({unsupported_vect_spec,BadVectSpec})
					   end
				   end
				   ||VectSpec<-VectSpecList],
	JoinLkpPipe=pipe_join_lookup:new(LookupFunList, ls, FormatPipe),
	expr(VExpr,[Expr|ExprPath],[JoinLkpPipe|PipeSequence],NewPg,PgtDictNew,_Id=make_ref()).

rsexpr({?RS_TAG,Iter}=_Expr,ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id) 
  	when is_record(Iter, iter) ->
	rsexpr({?RS_TAG,Iter,node()},ExprPath,PipeSequence,ParentPg,PgtDictAcc,Id);
rsexpr({?RS_TAG,Iter,Node}=_Expr,ExprPath,[ToPipe|_Other]=_PipeSequence,ParentPg,PgtDictAcc,Id) 
  	when is_record(Iter, iter) ->
 	vpump_work(Node,Iter,ExprPath,ToPipe,ParentPg,PgtDictAcc,Id).


%%---------------------------------------------------------
%% Local Functions
%%---------------------------------------------------------
vpump_work(Node,Iter,_ExprPath,ToPipe,ParentPg,PgtDictAcc,Id) ->
 	Pump=pump:new(Iter, ToPipe),
 	Worker=pump:to_pgt_worker(Id,Pump,Node),
	dict:append(ParentPg, Worker, PgtDictAcc).

strip_sender_sign([]=_ExprPath,DestPipe) ->
	pipe_mapper:new(fun({_SendSign,RecList})->
						RecList
					end, DestPipe);
strip_sender_sign([Head|_Tail]=_ExprPath,DestPipe) ->
	case element(1,Head) of
				?JOIN_TAG ->
					DestPipe;
				_ ->
					strip_sender_sign([],DestPipe)
	end.

route_snap_to_nodes(NRouteSnap) ->
	gb_sets:to_list(
	  lists:foldl(fun(Route,AccIn)->
						  Node=?ROUTE2UPNODE(Route),
						  gb_sets:add(Node, AccIn)
				  end, 
				  gb_sets:new(), 
				  gb_trees:values(NRouteSnap))).

make_pgt(PgtDict,{pg,Id,Name}=RootPg) ->
	{PgProtos,Workers}=case dict:find(RootPg, PgtDict) of
		error ->
			{[],[]};
		{ok,ChildrenList} when is_list(ChildrenList) ->
			{[Pg||Pg<-ChildrenList,element(1,Pg)==pg],[Wrk||Wrk<-ChildrenList,is_record(Wrk, worker)]}
	end,
	#pg{id=Id,name=Name,pgs=[make_pgt(PgtDict,Pg)||Pg<-PgProtos],workers=Workers}.

	
explain(Pg) ->
	pgt:explainex(Pg,fun(Tab,{Node,Module,Fun,_Args})->
							 lists:append([
										   io_lib:format(Tab++"Node=~p~n",[Node]),
							 			   io_lib:format(Tab++"Module=~p~n",[Module]),
							 			   io_lib:format(Tab++"Fun=~p~n",[Fun])])
				  end).


%%---------------------------------------------------------
%% Test Cases
%%---------------------------------------------------------	
test(map_rs) ->
	V1=vect_mnesia:new_mem_bag(ages1, [fname_lname,age]),
	Rs={?RS_TAG,iter:from_list([
									 {"Vasya","Ivanov",23},
									 {"Dima","Pupkin",27},
									 {"Masha","Bublikova",32},
									 {"Katya","Shkvarkina",19},
									 {"Yulya","Sharikova",34},
									 {"Vasya","Bukin",14},
									 {"Ostap","Bender",36},
									 {"Yuriy","Dolgorukiy",44},
									 {"Chingiz","Han",50}
									])},
	MapExpr1 = {?MAP_TAG, 
				Rs , 
				fun({Fn,Ln,Age}=_V)-> {ages1,{Fn,Ln},Age} end},
	eval(MapExpr1, pipe_vect:new(V1));

test(map_join) ->
	%load_ages_hobbies(),
	V1=vect_mnesia:new_mem_bag(ages, [id,age]),
	V2=vect_mnesia:new_mem_bag(hobbies, [id,hobby]),
	V3=vect_mnesia:new_mem_bag(j1, [fname,lname,age,hobby]),
	MapExpr1 = {?VECT_TAG,V1},
	MapExpr2 = {?VECT_TAG,V2},
	JoinExpr = {?JOIN_TAG,
				[MapExpr1,MapExpr2],
				fun([{ages,{Fn,Ln},Age},{hobbies,{Fn,Ln},Hobby}])-> 
						{j1,Fn,Ln,Age,Hobby};
				   (X) ->
						exit({bag_out_format,X})
				end },
	eval(JoinExpr, pipe_console:new()).

load_ages_hobbies() ->
	map_from_file("../test-data/ages.tup",vect_mnesia:new_mem_bag(ages, [id,age]),
				  fun({X,Y,Z})->
						  {ages,{X,Y},Z}
				  end),
	map_from_file("../test-data/hobbies.tup",vect_mnesia:new_mem_bag(hobbies, [id,hobby]),
				  fun({X,Y,Z})->
						  {hobbies,{X,Y},Z}
				  end).


map_from_file(Filename,Vect,RecFun) ->
	{ok,[Tuples]}=file:consult(Filename),
	Rs={?RS_TAG,iter:from_list(Tuples)},
	MapExpr1 = {?MAP_TAG, 
				Rs , 
				RecFun},
	pgt:new(eval(MapExpr1, pipe_vect:new(Vect))).

